package com.linkstec.loader;

import java.sql.ResultSet;
import java.sql.SQLException;
import java.text.SimpleDateFormat;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.Calendar;
import java.util.Date;
import java.util.HashMap;
import java.util.List;
import java.util.Map;

import org.apache.commons.lang.CharUtils;
import org.apache.commons.lang.StringUtils;
import org.apache.logging.log4j.LogManager;
import org.apache.logging.log4j.Logger;
import org.bson.Document;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.serializer.SerializerFeature;
import com.linkstec.mongo.MongoDbUtil;
import com.linkstec.redis.RedisUtil;
import com.mongodb.client.MongoCollection;
import com.mongodb.client.model.Filters;
import com.mongodb.client.result.DeleteResult;

import redis.clients.jedis.Jedis;
import redis.clients.jedis.Pipeline;

/**
 * 零编码缓存加载
 * @author CP_liuyonglu
 *
 */
public class LoaderEngine extends Loader {
	private static final Logger logger = LogManager.getLogger();

//	private String baseSql = "select * from MOT_LOADER_PROPERTIES WHERE LOADERID = ";
	private List<Document> documentList = new ArrayList<Document>();
	private MongoCollection<Document> collection = null;
	private static SimpleDateFormat sdf = new SimpleDateFormat("yyyyMMdd");
	static String storeType = null; // redis,mongo
	static String redisType = null; // String ,List , Set， Hash
	static String[] redisKeys = null;// 
	static String redisHashKey = null;
	static String[] redisValues = null;
	static Map<String,String> redisValuesMap = new HashMap<String,String>();
	static String sql = null;
	static int expireTime = 0;
	static int datePrefix = 0;
	static String valueType = null;

	/**
	 *
	 */

	public LoaderEngine(String loaderId, String loaderDesc, String sqlId) {
		super(loaderId, loaderDesc, sqlId);
	}

	@Override
	public void doLoad() throws Exception {
		/*ResultSet rs1 = Dao.getInstance().query(baseSql + "'" + loaderId + "'");

		while (rs1.next()) {
			storeType = rs1.getString("STORE_TYPE");
			redisType = rs1.getString("REDIS_TYPE");
			redisKeys = rs1.getString("REDIS_KEYS") == null ? null : rs1
					.getString("REDIS_KEYS").split(",");
			redisHashKey = rs1.getString("REDIS_HASH_KEY");
			redisValues = rs1.getString("REDIS_VALUES").split(",");
			sql = rs1.getString("SQL");
			expireTime = rs1.getInt("EXPIRE_TIME");
			datePrefix = rs1.getInt("DATE_PREFIX");
			valueType = rs1.getString("VALUE_TYPE");
			for (String values : redisValues) {
				redisValuesMap.put(values, fieldToProperty(values,false));
			}
		}*/

		ResultSet rs = Dao.getInstance().query(sql);
		long start = System.currentTimeMillis();
		try {
			loadCount = 0;

			if ("redis".equals(storeType)) {
				if ("Set_Batch".equals(redisType)) {
					//没有key字段，key不变
					 if (Arrays.asList(redisKeys).contains("-")) {
						 dealRsBatch(rs);
					 } else {
						 dealRsPipe(rs);
					 }
				}else if  ("ZSet_Batch".equals(redisType)) {
					dealRsZPipe(rs);
				} else if ("Bit_Batch".equals(redisType)) {
					dealRsPipeBit(rs);
				} else {
					dealRsRedis(rs);
				}
			} else {
				
				if (expireTime < 0) {
					if (null == collection) {
						collection = MongoDbUtil.getMongoCollection("MOT", loaderId);
					}
					Calendar calendar = Calendar.getInstance();
					calendar.add(Calendar.DATE, expireTime);
					String strToDay = sdf.format(calendar.getTime());
					//删除合并数据
					logger.info("开始删除" + strToDay + "前的数据");
					DeleteResult stockAnnResult = collection.deleteMany(Filters.lte("toDay", strToDay));
					logger.info("删除的条数是：" + stockAnnResult.getDeletedCount());
				}
				
				while (rs.next()) {
					
					Document document = new Document();
					
					for (String values : redisValues) {
						document.put(redisValuesMap.get(values), rs.getString(values));
					}
					document.put("toDay", sdf.format(new Date()));
					document.put("createTime", new Date());
					documentList.add(document);
					if(documentList.size() >= 1000){
						if (null == collection) {
							collection = MongoDbUtil.getMongoCollection("MOT", loaderId);
						}
						collection.insertMany(documentList);
						logger.info("成功插入了" + documentList.size() + "条数据！");
						
						documentList = null;
						documentList = new ArrayList<Document>();
					}
					
					loadCount++;
				}
				
				if(documentList.size() > 0){
					if (null == collection) {
						collection = MongoDbUtil.getMongoCollection("MOT", loaderId);
					}
					collection.insertMany(documentList);
					logger.info("成功插入了" + documentList.size() + "条数据！");
					
					documentList = null;
					documentList = new ArrayList<Document>();
				}
			}
			
		} catch (SQLException e) {
			logger.info("", e);
		} finally {
			// 资源关闭
			if (rs != null) {
				rs.close();
			}
		}
		costTime = System.currentTimeMillis() - start;
	}

	@Override
	public void beforeLoadData() throws Exception {
		super.beforeLoadData();
	}

	@Override
	public String afterLoadData() throws Exception {
		return super.afterLoadData();
	}
	/**
	 * 字段转换成对象属性 例如：user_name to userName
	 * @param field
	 * @param upcaseFirstChar
	 * @return
	 */
	public static String fieldToProperty(String field,boolean upcaseFirstChar) {
		if (null == field) {
			return "";
		}
		char[] chars = field.toLowerCase().toCharArray();
		StringBuffer sb = new StringBuffer();
		for (int i = 0; i < chars.length; i++) {
			char c = chars[i];
			if (c == '_') {
				int j = i + 1;
				if (j < chars.length) {
					sb.append(StringUtils.upperCase(CharUtils.toString(chars[j])));
					i++;
				}
			} else {
				if(upcaseFirstChar && i == 0){
					sb.append(StringUtils.upperCase(CharUtils.toString(c)));
				}else{
					sb.append(c);
				}

			}
		}
		return sb.toString();
	}

	/**
	 * 加载进redis
	 * @param rs
	 * @throws Exception 
	 */
	public void dealRsRedis(ResultSet rs) throws Exception {
		while (rs.next()) {
			
			String key = loaderId;
			if (1 == datePrefix) {
				key = key + "_" + sdf.format(new Date());
			}
			for (String keys : redisKeys) {
				key = key + "_" + rs.getString(keys);
			}
			
			String value = null;
			Map<String, Object> templetInfo = new HashMap<String, Object>();
			if ("String".equals(valueType)) {
				for (String values : redisValues) {
					value = rs.getString(values);
				}
			} else {
				for (String values : redisValues) {
					templetInfo.put(redisValuesMap.get(values), rs.getString(values));
				}
				value = JSON.toJSONString(templetInfo, SerializerFeature.WriteMapNullValue);
			}
			
			boolean flag = false;
			if (expireTime > 0){
				flag = RedisUtil.exists(key);
			}
			if ("String".equals(redisType)) {
				RedisUtil.set(key, value);
			} else if ("Set".equals(redisType)) {
				RedisUtil.addSet(key, value);
			} else if ("Hash".equals(redisType)) {
				RedisUtil.hSet(key, rs.getString(redisHashKey), value);
			} else if ("List".equals(redisType)) {
				RedisUtil.addList(key, value);
			}
			
			//超时时间
			if (expireTime > 0 && !flag)
				RedisUtil.expired(key, expireTime);
			
			loadCount++;
		}
	}

	/**
	 * 批量加载进redis
	 * @param rs
	 * @throws Exception
	 */
	public void dealRsBatch(ResultSet rs) throws Exception {
		List<String> list = new ArrayList<String>();
		Jedis jedis = RedisUtil.getJedis();
		String key = "SMOT_" + loaderId;
		if (1 == datePrefix) {
			key = key + "_" + sdf.format(new Date());
		}
		while (rs.next()) {
			//批量时候无效掉
			/*for (String keys : redisKeys) {
				key = key + "_" + rs.getString(keys);
			}*/
			
			String value = null;
			Map<String, Object> templetInfo = new HashMap<String, Object>();
			if ("String".equals(valueType)) {
				for (String values : redisValues) {
					value = rs.getString(values);
				}
			} else {
				for (String values : redisValues) {
					templetInfo.put(redisValuesMap.get(values),
							rs.getString(values));
				}
				value = JSON.toJSONString(templetInfo, SerializerFeature.WriteMapNullValue);
			}
			
			boolean flag = false;
			if (expireTime > 0){
				flag = RedisUtil.exists(key);
			}
			list.add(value);
			if (loadCount % 1000 == 0) {
				String[] custIdArr = list.toArray(new String[list.size()]);
				jedis.sadd(key, custIdArr);
				list.clear();
				logger.info("加载数量：" + loadCount);
			}
			//超时时间
			if (expireTime > 0 && !flag)
				RedisUtil.expired(key, expireTime);
			loadCount++;
		}
		if (list.size() > 0) {
			String[] custIdArr = list.toArray(new String[list.size()]);
			jedis.sadd(key, custIdArr);
			list.clear();
			logger.info("加载数量：" + loadCount);
		}
	}
	/**
	 * 批量加载进redis
	 * @param rs
	 * @throws Exception
	 */
	public void dealRsPipe(ResultSet rs) throws Exception {
		Jedis jedis = RedisUtil.getJedis();
		Pipeline pipeline = jedis.pipelined();
		while (rs.next()) {
			String key = "SMOT_" + loaderId;
			if (1 == datePrefix) {
				key = key + "_" + sdf.format(new Date());
			}

			for (String keys : redisKeys) {
				key = key + "_" + rs.getString(keys);
			}
			
			String value = null;
			Map<String, Object> templetInfo = new HashMap<String, Object>();
			if ("String".equals(valueType)) {
				for (String values : redisValues) {
					value = rs.getString(values);
				}
			} else {
				for (String values : redisValues) {
					templetInfo.put(redisValuesMap.get(values),
							rs.getString(values));
				}
				value = JSON.toJSONString(templetInfo, SerializerFeature.WriteMapNullValue);
			}
			
			pipeline.sadd(key, value);
			//超时时间
			if (expireTime > 0){
				pipeline.expire(key, expireTime);
			}
			if (loadCount % 1000 == 0) {
				pipeline.sync();
				logger.info("加载数量：" + loadCount);
			}
			loadCount++;
		}
		pipeline.sync();
		logger.info("加载数量：" + loadCount);
	}
	/**
	 * 批量加载进redis
	 * @param rs
	 * @throws Exception
	 */
	public void dealRsZPipe(ResultSet rs) throws Exception {
		Jedis jedis = RedisUtil.getJedis();
		Pipeline pipeline = jedis.pipelined();
		while (rs.next()) {
			String key = "SMOT_" + loaderId;
			if (1 == datePrefix) {
				key = key + "_" + sdf.format(new Date());
			}
			
			for (String keys : redisKeys) {
				key = key + "_" + rs.getString(keys);
			}
			
			String value = null;
			Map<String, Object> templetInfo = new HashMap<String, Object>();
			if ("String".equals(valueType)) {
				for (String values : redisValues) {
					value = rs.getString(values);
				}
			} else {
				for (String values : redisValues) {
					templetInfo.put(redisValuesMap.get(values),
							rs.getString(values));
				}
				value = JSON.toJSONString(templetInfo, SerializerFeature.WriteMapNullValue);
			}
			
			//redisHashKey 作为zset的score
			pipeline.zadd(key,rs.getDouble(redisHashKey), value);
			//超时时间
			if (expireTime > 0){
				pipeline.expire(key, expireTime);
			}
			if (loadCount % 1000 == 0) {
				pipeline.sync();
				logger.info("加载数量：" + loadCount);
			}
			loadCount++;
		}
		pipeline.sync();
		logger.info("加载数量：" + loadCount);
	}
	/**
	 * 批量加载进redis
	 * @param rs
	 * @throws Exception
	 */
	public void dealRsPipeBit(ResultSet rs) throws Exception {
		Jedis jedis = RedisUtil.getJedis();
		Pipeline pipeline = jedis.pipelined();
		while (rs.next()) {
			String key = "SMOT_" + loaderId;
			if (1 == datePrefix) {
				key = key + "_" + sdf.format(new Date());
			}
			//批量时候无效掉
			/*for (String keys : redisKeys) {
				key = key + "_" + rs.getString(keys);
			}*/
			
			long value = 0L;
			for (String values : redisValues) {
				value = rs.getLong(values);
			}
			
			//超时时间
			if (expireTime > 0){
				pipeline.expire(key, expireTime);
			}
			pipeline.setbit(key, value, true);
			if (loadCount % 1000 == 0) {
				pipeline.sync();
				logger.info("加载数量：" + loadCount);
			}
			loadCount++;
		}
		pipeline.sync();
		logger.info("加载数量：" + loadCount);
	}
	public static void main(String[] args) {
		
		Map<String, String> cfg = new HashMap<String, String>();
		cfg.put("keyNameSpace", "SMOT_");
		cfg.put("redisHostAndPortSingle", "10.0.209.232:19000,10.0.209.233:19000,10.0.209.234:19000");
//		cfg.put("redisHostAndPortSingle", "10.0.209.234:19000");
		cfg.put("passWord", "linkage@12345");
		RedisUtil.setConfig(cfg);
		
		String loaderId = "loaderId";
		String loaderDesc = "loaderDesc";
		
		storeType = "redis";
		redisType = "Set_Batch";
		redisKeys = "seccode".split(",");
		redisHashKey = "-";
		redisValues = "custno".split(",");
		sql = "SELECT T2.STOCK_CODE as seccode,T1.ENO as custno FROM ogg.T_GP T1 INNER JOIN ogg.t_Gp_Stock T2 ON T2.GP_ID = T1.ID";
		expireTime = 0;
		datePrefix = 0;
		valueType = "String";
		for (String values : redisValues) {
			redisValuesMap.put(values, fieldToProperty(values,false));
		}
		
		LoaderEngine loaderEngine = new LoaderEngine(loaderId,
				loaderDesc, null);
		try {
			loaderEngine.beforeLoadData();
			loaderEngine.doLoad();
			String afterMsg = loaderEngine.afterLoadData();
			logger.info(afterMsg);
		} catch (Exception e) {
			e.printStackTrace();
		}
	}
}